package com.learn.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * @create: 2023-04-17 15:54
 * @author: Mr.Du
 * --------------
 * @notes:
 **/
public class StreamWordCount {
    public static void main(String[] args) throws Exception {
        //构建流处理运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //构建socket流数据源
        DataStreamSource<String> lines = env.socketTextStream("node1", 9999);

        //数据处理
        lines.flatMap((String line, Collector<String> out) -> {
                    Arrays.stream(line.split(" ")).forEach(out::collect);
        }).returns(Types.STRING) //lambda实现一定要加returns
                .map((MapFunction<String, Tuple2<String, Integer>>) word -> Tuple2.of(word,1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .sum(1).print();

        env.execute("StreamWordCount");
    }
}
